#include "metanet_mq.h"

#include <assert.h>
#include <string.h>

#include "metanet_malloc.h"

#define DEFAULT_QUEUE_SIZE 64
#define MAX_GLOBAL_MQ 0x10000

#define MQ_IN_GLOBAL 1
#define MQ_OVERLOAD 1024

static struct metanet_global_message_queue *Q = NULL;

void 
metanet_globalmq_push(struct metanet_message_queue *queue) {
    struct metanet_global_message_queue *q = Q;
    assert(queue->next == NULL);
    if (q->tail) {
        q->tail->next = queue;
        q->tail = queue;
    } else {
        q->head = queue;
        q->tail = queue;
    }
}

void 
metanet_globalmq_push_head(struct metanet_message_queue *queue) {
    struct metanet_global_message_queue *q = Q;
    assert(queue->next == NULL);
    if (q->head) {
        queue->next = q->head;
        q->head = queue;
    } else {
        q->head = queue;
        q->tail = queue;
    }
}

struct metanet_message_queue * 
metanet_globalmq_pop(void) {
    struct metanet_global_message_queue *q = Q;
    struct metanet_message_queue *mq = q->head;
    if (mq) {
        q->head = mq->next;
        if (q->head == NULL) {
            assert(mq == q->tail);
            q->tail = NULL;
        }
        mq->next = NULL;
    }
    return mq;
}

struct metanet_message_queue *
metanet_mq_create(uint32_t handle) {
    struct metanet_message_queue *mq = metanet_malloc(sizeof(*mq));
    mq->handle = handle;
    mq->cap = DEFAULT_QUEUE_SIZE;
    mq->head = 0;
    mq->tail = 0;
    mq->overloaded = 0;
    mq->overloaded_threshold = MQ_OVERLOAD;
    mq->queue = metanet_malloc(sizeof(struct metanet_message) * mq->cap);
    mq->next = NULL;
    return mq;
}

static void
_release(struct metanet_message_queue *mq) {
    assert(mq->next == NULL);
    metanet_free(mq->queue);
    metanet_free(mq);
}

void 
metanet_mq_mark_release(struct metanet_message_queue *mq) {
    assert(mq->release == 0);
    mq->release = 1;
}

static void
_drop_queue(struct metanet_message_queue *mq, message_drop drop, void *ud) {
    struct metanet_message msg;
    while (!metanet_mq_pop(mq, &msg)) {
        drop(&msg, ud);
    }
    _release(mq);
}

void 
metanet_mq_release(struct metanet_message_queue *mq, message_drop drop, void *ud) {
    if (mq->release) {
        _drop_queue(mq, drop, ud);
    } else {
        metanet_globalmq_push(mq);
    }
}

uint32_t 
metanet_mq_handle(struct metanet_message_queue *mq) {
    return mq->handle;
}

int
metanet_mq_pop(struct metanet_message_queue *mq, struct metanet_message *msg) {
    assert(msg);
    assert(mq);
    int ret = 1;
    if (mq->head != mq->tail) {
        *msg = mq->queue[mq->head++];
        ret = 0;

        int head = mq->head;
        int tail = mq->tail;
        int cap = mq->cap;

        if (head >= cap) {
            mq->head = 0;
            head = 0;
        }
        int length = tail - head; // left size
        if (length < 0) {
            length += cap;
        }
        while (length > mq->overloaded_threshold) {
            mq->overloaded = length;
            mq->overloaded_threshold *= 2;
        }
    } else {
        mq->overloaded_threshold = MQ_OVERLOAD;
    }

    if (ret) {
        mq->in_global = 0;
    }

    return ret;
}

static void
metanet_message_queue_expand(struct metanet_message_queue *mq) {
    struct metanet_message *new_mq_queue = metanet_malloc(sizeof(struct metanet_message) * mq->cap * 2);
    int i;
    for (i =  0; i < mq->cap; ++i) {
        new_mq_queue[i] = mq->queue[(mq->head + i) % mq->cap];
    }
    mq->head = 0;
    mq->tail = mq->cap;
    mq->cap *= 2;

    metanet_free(mq->queue);
    mq->queue = new_mq_queue;
}

void 
metanet_mq_push(struct metanet_message_queue *mq, struct metanet_message *msg) {
    assert(msg);
    mq->queue[mq->tail] = *msg;
    if (++mq->tail >= mq->cap) {
        mq->tail = 0;
    }

    if (mq->head == mq->tail) {
        metanet_message_queue_expand(mq);
    }

    if (mq->in_global == 0) {
        mq->in_global = MQ_IN_GLOBAL;
        if (mq->priority) {
            metanet_globalmq_push_head(mq);
        } else {
            metanet_globalmq_push(mq);
        }
    }
}

void 
metanet_mq_init() {
    struct metanet_global_message_queue *queue = metanet_malloc(sizeof(*queue));
    memset(queue, 0, sizeof(*queue));
    Q = queue;
}

void
metanet_mq_exit() {
    if (Q != NULL) {
        metanet_free(Q);
        Q = NULL;
    }
}

int 
metanet_mq_length(struct metanet_message_queue *queue) {
    int head, tail, cap;

    head = queue->head;
    tail = queue->tail;
    cap = queue->cap;

    if (head <= tail) {
        return tail - head;
    } else {
        return tail - head + cap;
    }
}

int 
metanet_mq_overload(struct metanet_message_queue *queue) {
    if (queue->overloaded) {
        int overloaded = queue->overloaded;
        queue->overloaded = 0;
        return overloaded;
    }
    return 0;
}